package com.bytes32.classification.tree.dataset;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;

import com.bytes32.parsing.Parser;
import com.bytes32.parsing.ParsingException;

public final class Feature implements Writable{
	/**
	 * Attribute type (relates to value type)
	 * BINARY,LONG,DOUBLE,TEXT
	 */
	private AttributeType attributeType;
	
	/**
	 * Feature type,
	 * ordered, unordered, label, ignored
	 */
	private FeatureType featureType;
	
	/**
	 * The processor doing all the hard work
	 */
	@SuppressWarnings("rawtypes")
	private FeatureProcessor processor;
	
	@Override
	public void readFields(DataInput stream) throws IOException {
		this.parserId = WritableUtils.readVInt(stream);
		this.setFeatureType0(WritableUtils.readEnum(stream, FeatureType.class));
		this.attributeType = WritableUtils.readEnum(stream, AttributeType.class);
		if (this.processor!=null)
			this.processor.readFields(stream);
		
	}

	@Override
	public void write(DataOutput stream) throws IOException {
		WritableUtils.writeVInt(stream, this.parserId);
		WritableUtils.writeEnum(stream, this.featureType);
		WritableUtils.writeEnum(stream, this.attributeType);
		if (this.processor!=null)
			this.processor.write(stream);
	}


	/* helper methods */
	
	/**
	 * Increment the count 
	 * anything else will fail with a {@link InvalidFeatureValueException}
	 * @param value
	 * This works with {@link Number}, {@link String}, {@link Date} and byte[]
	 * @throws InvalidFeatureValueException 
	 */
	public void incrementCount(String value) throws InvalidFeatureValueException {
		if (value == null)
			throw new NullPointerException(" value should not be null ");
		if (this.processor == null)
			throw new IllegalStateException("Feature type is not set ");
		if (this.attributeType == null)
			throw new IllegalStateException("Attribute type is not set ");
		this.processor.handleString(value);
	}

	/**
	 * Increment the count 
	 * anything else will fail with a {@link InvalidFeatureValueException}
	 * @param value
	 * This works with {@link Number}, {@link String}, {@link Date} and byte[]
	 * @throws InvalidFeatureValueException 
	 */
	public void incrementCount(String value, Parser parser) throws InvalidFeatureValueException {
		if (value == null)
			throw new NullPointerException(" value should not be null ");
		if (this.processor == null)
			throw new IllegalStateException("Feature type is not set ");
		if (this.attributeType == null)
			throw new IllegalStateException("Attribute type is not set ");
		try {
			this.processor.handleString(parser.intoText(value));
		} catch (ParsingException e) {
			System.out.println(" Unable to parser even with selected parser ");
		}
	}	
	
	/**
	 * The Id of the winner parser
	 */
	private int parserId = -1;
	
	/**
	 * Increment the count 
	 * anything else will fail with a {@link InvalidFeatureValueException}
	 * @param value
	 * This works with {@link Number}, {@link String}, {@link Date} and byte[]
	 * @throws InvalidFeatureValueException 
	 */
	public void incrementCount(String value, List<Parser> parsers) throws InvalidFeatureValueException {
		if (value == null)
			throw new NullPointerException(" value should not be null ");
		if (this.processor == null)
			throw new IllegalStateException("Feature type is not set ");
		if (this.attributeType == null)
			throw new IllegalStateException("Attribute type is not set ");
		/* use the winner parser to translate the value */
		Parser win = parsers.get(parserId);
		try {
			this.processor.handleString(win.intoText(value));
		} catch (ParsingException e) {
			/* FIXME: do something here but if the winning parser can't deal with this who can?
			 * TODO: we should at least notify of failed parsing attempts */
			System.out.println(" Unable to parser even with selected parser ");			
		}
	}
	
	
	public AttributeType getAttributeType() {
		return attributeType;
	}

	public void setAttributeType(AttributeType attributeType) {
		this.attributeType = attributeType;
	}

	public FeatureType getFeatureType() {
		return featureType;
	}
	
	/**
	 * Returns the count of the values or splits when the Feature is un-ordered
	 * @return
	 */
	public long getCount(){
		return this.processor.getCount();
	}

	private void setFeatureType0(FeatureType featureType){
		this.featureType = featureType;
		switch (this.featureType) {
		case ORDERED:
			this.processor = new OrderedProcessor();
			break;
		case UNORDERED:
			this.processor = new UnOrderedProcessor();
			break;
		case UNKNOWN:
			this.processor = new DefaultFeatureProcessor();
			break;
		}		
	}
	
	/**
	 * Sets the feature type as well as the {@link FeatureProcessor}
	 * this can be either an {@link OrderedProcessor} or {@link UnOrderedProcessor}
	 * @param featureType
	 *  ORDERED or UNORDERED
	 */
	public void setFeatureType(FeatureType featureType) {
		this.setFeatureType0(featureType);
	}

	/**
	 * Returns the processor
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	public FeatureProcessor getProcessor() {
		return processor;
	}
	
	/**
	 * Merge this feature with another feature and
	 * return a result
	 * 
	 * Only successful if the features are
	 * of the same type
	 * @param feature
	 * @throws IllegalStateException when the features
	 * are not ordered or un-ordered 
	 */
	public void merge(Iterator<Feature> feature) {
		switch (this.featureType) {
		case ORDERED:
			mergeOrdered(feature);
			break;
		case UNORDERED:
			mergeUnOrdered(feature);
			break;
		default:
			throw new IllegalStateException(" Feature can be ordered or un-ordered ");
		}
		
	}
	
	private void mergeOrdered(final Iterator<Feature> buffer){
		OrderedProcessor temp = (OrderedProcessor)this.processor;
		temp.merge(new Iterator<OrderedProcessor>() {
			@Override
			public boolean hasNext() { return buffer.hasNext();}
			public OrderedProcessor next() {
				/* the calling method should make sure the feature types are the same */
				return (OrderedProcessor)buffer.next().processor;
			}
			public void remove() { throw new UnsupportedOperationException(" Not supported "); }
		});
	}
	
	private void mergeUnOrdered(final Iterator<Feature> buffer){
		UnOrderedProcessor temp = (UnOrderedProcessor)this.processor;
		temp.merge(new Iterator<UnOrderedProcessor>() {
			@Override
			public boolean hasNext() { return buffer.hasNext();}
			public UnOrderedProcessor next() {
				/* the calling method should make sure the feature types are the same */
				return (UnOrderedProcessor)buffer.next().processor;
			}
			public void remove() { throw new UnsupportedOperationException(" Not supported "); }
		});
		
	}
	
	/**
	 * Optimize the output before writing
	 * into the OutputStream
	 */
	public void compact(){
		this.processor.compact();
	}
	
	@Override
	public String toString() {
		return "[ "+this.processor.toString()+"]";
	}

	/**
	 * Sets the id of the successful parser 
	 * @param parserId
	 */
	public void setParserId(int parserId) {
		this.parserId = parserId;
	}
	
}
